use chrono::prelude::*;
use rumqttc::{AsyncClient, Event, MqttOptions, Outgoing, Packet, QoS};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::{task, time};
use serde_json::{Value};
use libsystemd::daemon::{self, NotifyState};

const MAX_RETRY_TIME: u64 = 60_000;

#[tokio::main]
async fn main() {
    let mut mqttoptions = MqttOptions::new("rumqtt", "127.0.0.1", 1883);
    mqttoptions.set_keep_alive(Duration::from_secs(5));

    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
    client
        .subscribe("hello/msg", QoS::AtMostOnce)
        .await
        .unwrap();

    let (tx, mut rx) = mpsc::channel(32);

    task::spawn(async move {
        let mut delay_ms = 100;
        loop {
            let result = eventloop.poll().await;

            // let time_stamp: DateTime<Local> = Local::now();
            match &result {
                Ok(Event::Incoming(Packet::ConnAck(_))) => {
                    tx.send(String::from("ReSub")).await.unwrap();
                    let now: DateTime<Local> = Local::now();
                    println!("ReSubscribe at {}", now);
                }
                Ok(Event::Incoming(Packet::Publish(p))) => {
                    // println!("Incoming = {:?}, {:?}", p.topic, p.payload);
                    let v: Value= serde_json::from_slice(&p.payload[..]).unwrap();
                    let id = (v["id"]).to_string();
                    let io = (v["io"]).to_string();
                    // let message = format!("{}+{}",id,io);
                    let message = id + "+" + &io;
                    tx.send(message).await.unwrap();
                }
                
                Ok(Event::Incoming(Packet::PingResp)) | 
                Ok(Event::Outgoing(Outgoing::PingReq)) => {}
                Ok(Event::Incoming(_packet)) => {
                    // let now: DateTime<Local> = Local::now();
                    // println!("Incoming = {:?} at {}", packet, now);
                }
                Ok(Event::Outgoing(_o)) => {
                    // let now: DateTime<Local> = Local::now();
                    // println!("Outgoing = {:?} at {}", o, now);
                }
                Err(e) => {
                    println!("Error = {:?}", e);
                    time::sleep(Duration::from_millis(delay_ms)).await;
                    println!("Delay = {}s", delay_ms);
                }
            }

            match &result {
                Ok(_) => {
                    delay_ms = 100;
                }
                Err(_) => {
                    if delay_ms < MAX_RETRY_TIME {
                        delay_ms = 2 * delay_ms;
                    } else {
                        delay_ms = MAX_RETRY_TIME;
                    }
                }
            }
        }
    });

    let _ = daemon::notify(true, &[NotifyState::Ready]).expect("notify failed");

    while let Some(message) = rx.recv().await {
        if message == "ReSub" {
            client
            .subscribe("hello/msg", QoS::AtMostOnce)
            .await
            .unwrap();
        }else{
            client
            .publish("hello/cmd", QoS::AtLeastOnce, false, message)
            .await
            .unwrap();
        }
        
    }
}
